package cn.doitedu.cn.doitedu.rtdw.data_sync;

import cn.doitedu.cn.doitedu.rtdw.ddl.CatalogDatabaseSwitch;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Job03_CmsVideoTable2Hbase {
    public static void main(String[] args) {

        // 编程环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:/d:/ckpt");
        env.setParallelism(1);

        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        // 切换catalog为hiveCatalog
        CatalogDatabaseSwitch.switchCatalogDatabase("doit40-hive", "doit40_rtdw", tenv);

        // 建表，映射mysql的视频信息源表
        tenv.executeSql("drop table if exists cms_video_mysql");
        tenv.executeSql(
                " create table cms_video_mysql(     " +
                        "    id int,                            " +
                        " 	 video_name string,                    " +
                        " 	 video_type string,                    " +
                        " 	 video_album string,                   " +
                        " 	 video_author string,                  " +
                        " 	 video_timelong bigint,                " +
                        "    PRIMARY KEY (id) NOT ENFORCED        " +
                        " ) with (                            "+
                        " 'connector' = 'mysql-cdc',          "+
                        " 'hostname' = 'doitedu'   ,          "+
                        " 'port' = '3306'          ,          "+
                        " 'username' = 'root'      ,          "+
                        " 'password' = 'root'      ,          "+
                        " 'database-name' = 'realtimedw',     "+
                        " 'table-name' = 'cms_video'         "+
                        " )                                   "
        );


        // 建表，映射hbase中的目标表
        tenv.executeSql("drop table if exists dim_video_hbase");
        tenv.executeSql(
                " create table dim_video_hbase(              " +
                        "   id int,                            " +
                        " 	f row<                                " +
                        " 	video_name string,                    " +
                        " 	video_type string,                    " +
                        " 	video_album string,                   " +
                        " 	video_author string,                  " +
                        " 	video_timelong bigint                 " +
                        " 	>,                                    " +
                        " 	PRIMARY KEY (id) NOT ENFORCED    " +
                        " ) with (                                " +
                        "  'connector' = 'hbase-2.2',             " +
                        "  'table-name' = 'dim_video',            " +
                        "  'zookeeper.quorum' = '192.168.77.88:2181'    " +
                        " )                                       "
        );

        tenv.executeSql(
                " insert into dim_video_hbase " +
                        " SELECT                      " +
                        "     id                      " +
                        "     ,row(	                  " +
                        "     video_name              " +
                        "     ,video_type             " +
                        "     ,video_album            " +
                        "     ,video_author           " +
                        "     ,video_timelong) as f   " +
                        " FROM cms_video_mysql        "
        );


    }


}
